package com.amigo.online.provider.kafka.sender;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations.OperationsCallback;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;


@Component
public class KafkaMessageSender {
	
   /* private KafkaMessageSender(){
    	
    }
	
	private static class SingletonInstance{
		private static final KafkaMessageSender INSTANCE =  new KafkaMessageSender();
	}
	
	public static KafkaMessageSender getInstance() {
		return SingletonInstance.INSTANCE;
	}*/

    private static Logger log = LoggerFactory.getLogger(KafkaMessageSender.class);

    @Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    private Gson gson = new GsonBuilder().create();

    /**
     * 发送失败日志
     *
     * @param topic
     * @param msg
     */
    public void failLogSend(String topic, Object msg) {

        //Message message = getNewMessage(msg);
        log.info("+++++++++++++++++++++  message = {}");
        ListenableFuture<SendResult<String, String>> result = kafkaTemplate.send(topic, gson.toJson(msg));
        result.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            //发送成功
            @Override
            public void onSuccess(SendResult<String, String> result) {

            }

            //发送失败
            @Override
            public void onFailure(Throwable ex) {
                log.error("日志发送失败,日志信息为:" + msg.toString());
            }

        });

        kafkaTemplate.partitionsFor("maxwell");

    }

    public void failLogSendBlock(String topic, Object msg) {

        //Message message = getNewMessage(msg);
        log.info("+++++++++++++++++++++  message = {}");
        try {
            kafkaTemplate.send(topic, gson.toJson(msg)).get(10, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            //  Auto-generated catch block
            e.printStackTrace();
        } catch (ExecutionException e) {
            //  Auto-generated catch block
            e.printStackTrace();
        } catch (TimeoutException e) {
            //  Auto-generated catch block
            e.printStackTrace();
        }


        kafkaTemplate.partitionsFor("maxwell");

    }

    public void sendToKafka(Object data) {
        final ProducerRecord<String, String> record;
    }
  
    
 /*   public <Boolean> boolean sendTransactionMessage() {
    	Boolean result = kafkaTemplate.executeInTransaction((OperationsCallback<String, String, Boolean>) t -> {
    		t.sendDefault("");
			return true;
    	});
    }*/
}
